package org.smarthings.mqtt;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.dna.mqtt.moquette.server.netty.NettyAcceptor;
import org.smarthings.Message;
import org.smarthings.MessageBroker;
import org.smarthings.Server;
import org.smarthings.Utils;

public class MqttServerImpl implements Server, IMessaging {
	private static final MessageBroker broker = MessageBroker.getInstance();
	private List<String> subscriptions = new ArrayList<String>();
	private SimpleMessaging messaging;
	private NettyAcceptor acceptor;
	private MqttClientImpl client;

	public void start(String host, int port) {
		try {
			Properties props = new Properties();
			props.put("host", host);
			props.put("port", "" + port);
			props.put("password_file", "mqtt.passwords");
			messaging = SimpleMessaging.getInstance();
			messaging.init(props);
			acceptor = new NettyAcceptor();
			acceptor.initialize(this, props);
			client = new MqttClientImpl();
			client.start(host, port);
			System.out.println(MqttFactory.PROTOCOL + " SERVER STARTED");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void stop() {
		messaging.stop();
		acceptor.close();
		subscriptions.clear();
		System.out.println(MqttFactory.PROTOCOL + " SERVER STOPPED");
	}

	public String getProtocol() {
		return MqttFactory.PROTOCOL;
	}

	public void update(String uri) {
		if (subscriptions.contains(uri)) {
			Message message = broker.get(uri, "application/json");
			client.post(uri + ".update", message);
		}
	}

	public void disconnect(ServerChannel session) {
		messaging.disconnect(session);
	}

	public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
		if (msg instanceof PublishMessage) {
			PublishMessage pub = (PublishMessage) msg;
			byte[] payload = pub.getPayload().array();
			String topic = pub.getTopicName();
			String uri = Utils.normalizeUri(topic);

			if (!uri.endsWith(".update")) {
				Message message = Message.createSimpleMessage(payload, MqttFactory.PROTOCOL, uri, "application/json");
				broker.post(message);
			} else {
				pub.setTopicName(uri.substring(0, uri.length() - 7));
			}
		}

		if (msg instanceof SubscribeMessage) {
			SubscribeMessage sub = (SubscribeMessage) msg;
			List<Couple> couples = sub.subscriptions();
			for (Couple couple : couples) {
				String topic = couple.getTopic();
				subscriptions.add(topic);
			}
		}

		if (msg instanceof UnsubscribeMessage) {
			UnsubscribeMessage unsub = (UnsubscribeMessage) msg;
			List<String> topics = unsub.topics();
			for (String topic : topics) {
				subscriptions.remove(topic);
			}
		}

		messaging.handleProtocolMessage(session, msg);
	}

	public void lostConnection(String clientID) {
		messaging.lostConnection(clientID);
	}
}
